package flink_table

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors.{FileSystem, OldCsv, Schema}
//隐式转换必须
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
object demo1 {
    def main(args: Array[String]): Unit = {
        //流处理环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)

        //创建table执行环境
        val tableEnv = StreamTableEnvironment.create(env)

        //连接外部系统读取数据
        val filepath:String = "F:\\我的世界\\hive\\hive--2\\resource\\stu.txt"
        tableEnv.connect(new FileSystem().path(filepath))
            .withFormat(new OldCsv())   // 定义反序列化的格式
            .withSchema(new Schema()    // 定义schema
                .field("id",DataTypes.INT())
                .field("name",DataTypes.STRING())
                .field("sex",DataTypes.STRING())
                .field("age",DataTypes.DOUBLE())
                .field("subject",DataTypes.STRING())
            ).createTemporaryTable("student")

        val table: Table = tableEnv.from("student")
        table.toAppendStream[(Int,String,String,Double,String)].print()
        env.execute("f_sql")
    }
}
